--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 3bc3494a58f2247bd329485110c1e15932e5fe1d
Parents : f335044
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-03-06T15:50:30-06:00
Add extensive tests for notification reliability and LXMF field hardening
Changes
2 files changed, 539 insertions(+), 0 deletions(-)
Diff
diff --git a/tests/backend/test_notifications.py b/tests/backend/test_notifications.py
index d1df2c51..f85e7786 100644
--- a/tests/backend/test_notifications.py
+++ b/tests/backend/test_notifications.py
@@ -342,3 +342,239 @@ def test_notification_spike_fuzzing(db, num_notifs):
notifications = db.misc.get_notifications(limit=num_notifs)
assert len(notifications) == num_notifs
+
+
+# ---------------------------------------------------------------------------
+# Extensive notification reliability tests
+# ---------------------------------------------------------------------------
+
+
+class TestNotificationReliability:
+ """Comprehensive tests to verify notifications are accurate, reliable,
+ and never produce false positives."""
+
+ def test_unread_count_matches_actual_unread(self, db):
+ """Unread count must exactly match unviewed notifications."""
+ for i in range(10):
+ db.misc.add_notification(f"type{i}", f"h{i}", f"t{i}", f"c{i}")
+
+ assert db.misc.get_unread_notification_count() == 10
+
+ ids = [n["id"] for n in db.misc.get_notifications()]
+ db.misc.mark_notifications_as_viewed([ids[0], ids[1], ids[2]])
+ assert db.misc.get_unread_notification_count() == 7
+
+ db.misc.mark_notifications_as_viewed()
+ assert db.misc.get_unread_notification_count() == 0
+
+ def test_marking_empty_list_marks_all(self, db):
+ """Marking with an empty list (falsy) falls through to mark-all behavior."""
+ db.misc.add_notification("t", "h", "title", "content")
+ db.misc.mark_notifications_as_viewed([])
+ assert db.misc.get_unread_notification_count() == 0
+
+ def test_marking_nonexistent_ids_is_safe(self, db):
+ """Marking IDs that don't exist should not crash."""
+ db.misc.mark_notifications_as_viewed([99999, 88888, 77777])
+ assert db.misc.get_unread_notification_count() == 0
+
+ def test_double_mark_viewed_idempotent(self, db):
+ """Marking the same notification as viewed twice is safe."""
+ db.misc.add_notification("t", "h", "title", "content")
+ nid = db.misc.get_notifications()[0]["id"]
+ db.misc.mark_notifications_as_viewed([nid])
+ db.misc.mark_notifications_as_viewed([nid])
+ assert db.misc.get_unread_notification_count() == 0
+
+ def test_no_ghost_notifications_after_clear(self, db):
+ """After marking all as viewed, unread count must be zero and stay zero."""
+ for i in range(5):
+ db.misc.add_notification(f"type{i}", "h", "t", "c")
+ db.misc.mark_notifications_as_viewed()
+ assert db.misc.get_unread_notification_count() == 0
+ assert len(db.misc.get_notifications(filter_unread=True)) == 0
+
+ def test_interleaved_add_and_mark(self, db):
+ """Adding and marking interleaved must produce correct counts."""
+ db.misc.add_notification("a", "h1", "t1", "c1")
+ db.misc.add_notification("b", "h2", "t2", "c2")
+ nid1 = db.misc.get_notifications()[0]["id"]
+ db.misc.mark_notifications_as_viewed([nid1])
+
+ db.misc.add_notification("c", "h3", "t3", "c3")
+ assert db.misc.get_unread_notification_count() == 2
+
+ def test_limit_does_not_affect_unread_count(self, db):
+ """get_unread_notification_count is independent of query limit."""
+ for i in range(20):
+ db.misc.add_notification(f"t{i}", "h", "title", "content")
+ limited = db.misc.get_notifications(limit=5)
+ assert len(limited) == 5
+ assert db.misc.get_unread_notification_count() == 20
+
+ def test_missed_call_creates_exactly_one_notification(self, mock_app):
+ """A single missed call must produce exactly one notification."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ caller = MagicMock()
+ caller.hash = b"caller_hash_32_bytes_long_012345"
+ mock_app.telephone_manager.call_is_incoming = True
+ mock_app.telephone_manager.call_status_at_end = 4
+ mock_app.telephone_manager.call_start_time = time.time() - 5
+ mock_app.telephone_manager.call_was_established = False
+
+ mock_app.on_telephone_call_ended(caller)
+ notifications = mock_app.database.misc.get_notifications()
+ assert len(notifications) == 1
+ assert notifications[0]["type"] == "telephone_missed_call"
+
+ def test_established_call_creates_no_notification(self, mock_app):
+ """An established (answered) call must not create a missed-call notification."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ caller = MagicMock()
+ caller.hash = b"caller_hash_32_bytes_long_012345"
+ mock_app.telephone_manager.call_is_incoming = True
+ mock_app.telephone_manager.call_status_at_end = 6
+ mock_app.telephone_manager.call_start_time = time.time() - 60
+ mock_app.telephone_manager.call_was_established = True
+
+ mock_app.on_telephone_call_ended(caller)
+ notifications = mock_app.database.misc.get_notifications()
+ assert len(notifications) == 0
+
+ def test_outgoing_call_creates_no_missed_notification(self, mock_app):
+ """An outgoing call that ends without answer must not produce a missed-call notification."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ callee = MagicMock()
+ callee.hash = b"callee_hash_32_bytes_long_012345"
+ mock_app.telephone_manager.call_is_incoming = False
+ mock_app.telephone_manager.call_status_at_end = 2
+ mock_app.telephone_manager.call_start_time = time.time() - 10
+ mock_app.telephone_manager.call_was_established = False
+
+ mock_app.on_telephone_call_ended(callee)
+ notifications = mock_app.database.misc.get_notifications()
+ assert len(notifications) == 0
+
+ def test_voicemail_creates_exactly_one_notification(self, mock_app):
+ """A voicemail must create exactly one notification."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+ mock_app.on_new_voicemail_received("abc123", "User X", 30)
+ notifications = mock_app.database.misc.get_notifications()
+ assert len(notifications) == 1
+ assert notifications[0]["type"] == "telephone_voicemail"
+
+ def test_rapid_missed_calls_unique_notifications(self, mock_app):
+ """Multiple rapid missed calls from different callers produce separate notifications."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ for i in range(10):
+ caller = MagicMock()
+ caller.hash = f"caller_{i:032d}".encode()[:32]
+ mock_app.telephone_manager.call_is_incoming = True
+ mock_app.telephone_manager.call_status_at_end = 4
+ mock_app.telephone_manager.call_start_time = time.time() - 1
+ mock_app.telephone_manager.call_was_established = False
+ mock_app.on_telephone_call_ended(caller)
+
+ notifications = mock_app.database.misc.get_notifications()
+ assert len(notifications) == 10
+ assert all(n["type"] == "telephone_missed_call" for n in notifications)
+
+ def test_mixed_notification_types_correct_counts(self, mock_app):
+ """Mix of missed calls and voicemails produces correct per-type counts."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ for _ in range(3):
+ caller = MagicMock()
+ caller.hash = b"caller_hash_32_bytes_long_012345"
+ mock_app.telephone_manager.call_is_incoming = True
+ mock_app.telephone_manager.call_status_at_end = 4
+ mock_app.telephone_manager.call_start_time = time.time()
+ mock_app.telephone_manager.call_was_established = False
+ mock_app.on_telephone_call_ended(caller)
+
+ for _ in range(2):
+ mock_app.on_new_voicemail_received("vm_hash", "VmUser", 10)
+
+ notifications = mock_app.database.misc.get_notifications()
+ missed = [n for n in notifications if n["type"] == "telephone_missed_call"]
+ voicemails = [n for n in notifications if n["type"] == "telephone_voicemail"]
+ assert len(missed) == 3
+ assert len(voicemails) == 2
+ assert mock_app.database.misc.get_unread_notification_count() == 5
+
+ def test_mark_viewed_reduces_count_precisely(self, mock_app):
+ """Marking specific notifications as viewed reduces count by exactly that many."""
+ mock_app.database.misc.provider.execute("DELETE FROM notifications")
+
+ for i in range(5):
+ mock_app.database.misc.add_notification(f"t{i}", f"h{i}", f"t{i}", f"c{i}")
+
+ assert mock_app.database.misc.get_unread_notification_count() == 5
+
+ ids = [n["id"] for n in mock_app.database.misc.get_notifications()]
+ mock_app.database.misc.mark_notifications_as_viewed([ids[0], ids[2]])
+ assert mock_app.database.misc.get_unread_notification_count() == 3
+
+ def test_notification_ordering(self, db):
+ """Notifications should be returned in reverse chronological order (newest first)."""
+ import time as _time
+
+ for i in range(5):
+ db.misc.add_notification(f"type{i}", "h", f"title_{i}", f"content_{i}")
+ _time.sleep(0.01)
+
+ notifications = db.misc.get_notifications()
+ for j in range(len(notifications) - 1):
+ assert notifications[j]["timestamp"] >= notifications[j + 1]["timestamp"]
+
+
+@settings(deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture])
+@given(
+ num_add=st.integers(min_value=0, max_value=50),
+ num_mark=st.integers(min_value=0, max_value=50),
+)
+def test_add_mark_invariant(db, num_add, num_mark):
+ """Invariant: unread count == total - marked, never negative."""
+ db.provider.execute("DELETE FROM notifications")
+ for i in range(num_add):
+ db.misc.add_notification(f"t{i}", "h", "t", "c")
+
+ all_notifs = db.misc.get_notifications()
+ ids_to_mark = [n["id"] for n in all_notifs[:num_mark]]
+ if ids_to_mark:
+ db.misc.mark_notifications_as_viewed(ids_to_mark)
+
+ expected = max(0, num_add - len(ids_to_mark))
+ assert db.misc.get_unread_notification_count() == expected
+
+
+@settings(deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture])
+@given(
+ operations=st.lists(
+ st.tuples(
+ st.sampled_from(["add", "mark_one", "mark_all"]),
+ st.text(min_size=1, max_size=20),
+ ),
+ min_size=1,
+ max_size=30,
+ ),
+)
+def test_random_notification_operations(db, operations):
+ """Fuzz random sequences of add/mark operations; count must never go negative."""
+ db.provider.execute("DELETE FROM notifications")
+ for op, payload in operations:
+ if op == "add":
+ db.misc.add_notification("type", payload, "title", "content")
+ elif op == "mark_one":
+ notifs = db.misc.get_notifications(limit=1)
+ if notifs:
+ db.misc.mark_notifications_as_viewed([notifs[0]["id"]])
+ elif op == "mark_all":
+ db.misc.mark_notifications_as_viewed()
+
+ count = db.misc.get_unread_notification_count()
+ assert count >= 0
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index 6c336688..c1ad9eea 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -2196,3 +2196,306 @@ def test_lxmf_display_name_parsing_regression():
# None input
assert parse_lxmf_display_name(None, default_value="Fallback") == "Fallback"
+
+
+class TestLxmfFieldHardening:
+ """Tests for LXMF message field type safety and edge cases."""
+
+ def test_file_attachment_non_list(self, mock_app):
+ """File attachments field as non-list should not crash."""
+ mock_message = MagicMock()
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+ mock_message.content = b"test"
+ mock_message.title = b""
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.delivery_attempts = 1
+ mock_message.next_delivery_attempt = None
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_FILE_ATTACHMENTS: "not-a-list",
+ }
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ result = convert_lxmf_message_to_dict(mock_message)
+ assert isinstance(result, dict)
+
+ def test_file_attachment_item_wrong_type(self, mock_app):
+ """File attachment items that are not [name, data] pairs should not crash."""
+ mock_message = MagicMock()
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+ mock_message.content = b"test"
+ mock_message.title = b""
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.delivery_attempts = 1
+ mock_message.next_delivery_attempt = None
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_FILE_ATTACHMENTS: [42, None, "string", {"a": "b"}],
+ }
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ try:
+ convert_lxmf_message_to_dict(mock_message)
+ except (TypeError, IndexError, KeyError):
+ pass
+
+ def test_image_field_non_list(self, mock_app):
+ """Image field as unexpected type should not crash."""
+ mock_message = MagicMock()
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+ mock_message.content = b"test"
+ mock_message.title = b""
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.delivery_attempts = 1
+ mock_message.next_delivery_attempt = None
+ mock_message.get_fields.return_value = {
+ LXMF.FIELD_IMAGE: 42,
+ }
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ try:
+ convert_lxmf_message_to_dict(mock_message)
+ except (TypeError, IndexError, KeyError):
+ pass
+
+ @settings(
+ suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None
+ )
+ @given(
+ fields_data=st.dictionaries(
+ keys=st.integers(min_value=0, max_value=20),
+ values=st.one_of(
+ st.none(),
+ st.booleans(),
+ st.integers(),
+ st.text(max_size=200),
+ st.binary(max_size=200),
+ st.lists(st.text(max_size=50), max_size=5),
+ ),
+ max_size=10,
+ ),
+ )
+ def test_arbitrary_fields_do_not_crash(self, mock_app, fields_data):
+ """Fuzz arbitrary LXMF fields dict to ensure no crashes."""
+ mock_message = MagicMock()
+ mock_message.source_hash = os.urandom(16)
+ mock_message.destination_hash = os.urandom(16)
+ mock_message.hash = os.urandom(16)
+ mock_message.content = b"ok"
+ mock_message.title = b""
+ mock_message.incoming = True
+ mock_message.state = LXMF.LXMessage.DELIVERED
+ mock_message.method = LXMF.LXMessage.DIRECT
+ mock_message.progress = 1.0
+ mock_message.timestamp = 123456789.0
+ mock_message.rssi = -50
+ mock_message.snr = 10
+ mock_message.q = 100
+ mock_message.delivery_attempts = 1
+ mock_message.next_delivery_attempt = None
+ mock_message.get_fields.return_value = fields_data
+ from meshchatx.src.backend.lxmf_utils import convert_lxmf_message_to_dict
+
+ try:
+ convert_lxmf_message_to_dict(mock_message)
+ except (TypeError, IndexError, KeyError, ValueError):
+ pass
+
+
+class TestStrangerAttachmentBlocking:
+ """Tests for stranger attachment stripping logic."""
+
+ def _make_mock_message(self, source_hash=None, with_attachments=True):
+ mock_msg = MagicMock()
+ mock_msg.source_hash = source_hash or os.urandom(16)
+ mock_msg.destination_hash = os.urandom(16)
+ mock_msg.hash = os.urandom(16)
+ mock_msg.content = b"hello from stranger"
+ mock_msg.title = b""
+ mock_msg.incoming = True
+ mock_msg.state = LXMF.LXMessage.DELIVERED
+ mock_msg.method = LXMF.LXMessage.DIRECT
+ mock_msg.progress = 1.0
+ mock_msg.timestamp = 123456789.0
+ mock_msg.rssi = -50
+ mock_msg.snr = 10
+ mock_msg.q = 100
+ mock_msg.delivery_attempts = 1
+ mock_msg.next_delivery_attempt = None
+
+ fields = {}
+ if with_attachments:
+ fields[LXMF.FIELD_FILE_ATTACHMENTS] = [
+ [b"malware.exe", b"\x00" * 100],
+ ]
+ mock_msg.get_fields.return_value = fields
+ mock_msg.fields = fields
+ return mock_msg
+
+ def test_stranger_attachments_stripped_when_enabled(self, mock_app):
+ """Attachments from non-contacts should be stripped when setting is on."""
+ source_hash = os.urandom(16)
+ mock_msg = self._make_mock_message(source_hash=source_hash)
+
+ mock_app.config.block_attachments_from_strangers.get.return_value = True
+ mock_app.config.block_all_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+ call_kwargs = mock_app.db_upsert_lxmf_message.call_args
+ assert call_kwargs[1].get("attachments_stripped") is True or (
+ len(call_kwargs[0]) > 2 and call_kwargs[0][2] is True
+ )
+
+ def test_contact_attachments_not_stripped(self, mock_app):
+ """Attachments from contacts should NOT be stripped."""
+ source_hash = os.urandom(16)
+ mock_msg = self._make_mock_message(source_hash=source_hash)
+
+ mock_app.config.block_attachments_from_strangers.get.return_value = True
+ mock_app.config.block_all_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=True)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+ fields = mock_msg.get_fields()
+ assert LXMF.FIELD_FILE_ATTACHMENTS in fields
+
+ def test_setting_disabled_allows_stranger_attachments(self, mock_app):
+ """When setting is off, stranger attachments pass through."""
+ source_hash = os.urandom(16)
+ mock_msg = self._make_mock_message(source_hash=source_hash)
+
+ mock_app.config.block_attachments_from_strangers.get.return_value = False
+ mock_app.config.block_all_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+ fields = mock_msg.get_fields()
+ assert LXMF.FIELD_FILE_ATTACHMENTS in fields
+
+ def test_text_message_from_stranger_still_delivered(self, mock_app):
+ """Text-only messages from strangers are delivered normally."""
+ source_hash = os.urandom(16)
+ mock_msg = self._make_mock_message(
+ source_hash=source_hash, with_attachments=False
+ )
+
+ mock_app.config.block_attachments_from_strangers.get.return_value = True
+ mock_app.config.block_all_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+
+class TestBlockAllFromStrangers:
+ """Tests for the block-everything-from-strangers feature."""
+
+ def _make_mock_message(self, source_hash=None, with_attachments=False):
+ mock_msg = MagicMock()
+ mock_msg.source_hash = source_hash or os.urandom(16)
+ mock_msg.destination_hash = os.urandom(16)
+ mock_msg.hash = os.urandom(16)
+ mock_msg.content = b"hello from stranger"
+ mock_msg.title = b""
+ mock_msg.incoming = True
+ mock_msg.state = LXMF.LXMessage.DELIVERED
+ mock_msg.method = LXMF.LXMessage.DIRECT
+ mock_msg.progress = 1.0
+ mock_msg.timestamp = 123456789.0
+ mock_msg.rssi = -50
+ mock_msg.snr = 10
+ mock_msg.q = 100
+ mock_msg.delivery_attempts = 1
+ mock_msg.next_delivery_attempt = None
+ fields = {}
+ if with_attachments:
+ fields[LXMF.FIELD_FILE_ATTACHMENTS] = [[b"file.txt", b"data"]]
+ mock_msg.get_fields.return_value = fields
+ mock_msg.fields = fields
+ return mock_msg
+
+ def test_stranger_message_dropped_when_enabled(self, mock_app):
+ """Text message from a stranger is silently dropped when block_all is on."""
+ mock_msg = self._make_mock_message()
+ mock_app.config.block_all_from_strangers.get.return_value = True
+ mock_app.config.block_attachments_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_not_called()
+
+ def test_contact_message_delivered_when_block_all_enabled(self, mock_app):
+ """Messages from contacts pass through even when block_all is on."""
+ mock_msg = self._make_mock_message()
+ mock_app.config.block_all_from_strangers.get.return_value = True
+ mock_app.config.block_attachments_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=True)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
+
+ def test_stranger_with_attachments_dropped(self, mock_app):
+ """Message with attachments from stranger is dropped entirely, not just stripped."""
+ mock_msg = self._make_mock_message(with_attachments=True)
+ mock_app.config.block_all_from_strangers.get.return_value = True
+ mock_app.config.block_attachments_from_strangers.get.return_value = True
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_not_called()
+
+ def test_disabled_allows_stranger_messages(self, mock_app):
+ """When block_all is off, stranger messages are delivered."""
+ mock_msg = self._make_mock_message()
+ mock_app.config.block_all_from_strangers.get.return_value = False
+ mock_app.config.block_attachments_from_strangers.get.return_value = False
+ mock_app._is_contact = MagicMock(return_value=False)
+ mock_app.is_destination_blocked = MagicMock(return_value=False)
+ mock_app.check_spam_keywords = MagicMock(return_value=False)
+
+ mock_app.on_lxmf_delivery(mock_msg)
+ mock_app.db_upsert_lxmf_message.assert_called_once()
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────